package com.tarena.charging.order.mqtt;

import cn.tedu.charging.common.pojo.JsonResult;
import cn.tedu.charging.common.pojo.param.ChargingProcessParam;
import cn.tedu.charging.common.pojo.vo.ChargingProcessVO;
import com.alibaba.fastjson2.JSON;
import com.tarena.charging.order.constant.OrderStatusConstant;
import com.tarena.charging.order.constant.TopicConstant;
import com.tarena.charging.order.feign.CostClient;
import com.tarena.charging.order.pojo.dto.ChargingProcessMessageDto;
import com.tarena.charging.order.pojo.dto.ChargingResultMessageDto;
import com.tarena.charging.order.pojo.entity.ChargingProcessEntity;
import com.tarena.charging.order.pojo.po.BillFailPO;
import com.tarena.charging.order.pojo.po.BillSuccessPO;
import com.tarena.charging.order.websocket.WebSocketServer;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.BeanUtils;

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {
    private MqttClient client;
    private MqttContext mqttContext;
    public MqttConsumer(MqttContext context,MqttClient client){
        log.info("当前订单消费者被创建了");
        this.mqttContext = context;
        this.client = client;
    }
    /*
        定义链接emqx成功后 当前消费者需要订阅的主题
     */
    @Override public void connectComplete(boolean reconnect, String serverURI) {
        //当前订单可能订阅2个不同的主题 一个接收自检结果的主题 一个是充电进度主题
        log.info("MqttConsumer#connectComplete方法被调用了,表示连接成功," +
            "reconnect:{},serverURI:{}",reconnect,serverURI);
        try {
            log.info("连接成功,订单开始订阅充电桩反馈结果主题:{}", TopicConstant.CHARGING_RESULT);
            client.subscribe(TopicConstant.CHARGING_RESULT);
            log.info("连接成功,订单订阅充电桩反馈结果主题:{}成功",TopicConstant.CHARGING_RESULT);
            log.info("连接成功,订单开始订阅充电进度主题:{}", TopicConstant.CHARGING_PREGRESS);
            client.subscribe(TopicConstant.CHARGING_PREGRESS);
            log.info("连接成功,订单订阅充电进度主题:{}成功",TopicConstant.CHARGING_PREGRESS);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    @Override public void connectionLost(Throwable cause) {
        log.info("MqttConsumer#connectionLost方法被调用了,表示连接丢失",cause);
    }
    //订阅链接主题后 消息传递过来 的消费逻辑方法
    @Override public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("主题:{},收到消息:{}",topic,message.toString());
        //订单消费者 订阅了2个主题的 条件判断拆分 分别对待不同消息主题中的消息
        if (topic!=null&&topic.equals(TopicConstant.CHARGING_RESULT)){
            log.info("订单开始处理 充电桩自检结果");
            handleChargingResult(message);
        }else if (topic!=null&&topic.equals(TopicConstant.CHARGING_PREGRESS)){
            log.info("订单开始处理 充电桩充电进度结果");
            handleCharginProgress(message);
        }
    }

    //当监听主题 是充电进度主题的时候 会调用这个方法处理
    private void handleCharginProgress(MqttMessage message) {
        log.info("处理充电进度:{}",message);
        //1.解析一下这个参数转化成ChargingProcessDto
        ChargingProcessMessageDto cpmd=parseMessage(message);
        log.info("拿到充电桩 进度消息解析结果:{}",cpmd);
        //2.在消息对象非空的时候 我们可以调用计价服务 计算价格 1度电,1.5元 2度电 3元 上次充了5元 加上本次3元 计价返回8元
        ChargingProcessVO cost=billPriceCalc(cpmd);
        log.info("计算价钱完毕,最终计价结果:{}",cost);
        //3.存储最终结果 到es 保证 每次存储时序的 id 按照时间生成
        saveChargingProcessEntity(cpmd,cost);
        //4.websocket 同步推送给用户 实时进度数据.
        sendChargingProcess2User(cost);
        if (cpmd.getIsFull()){
            //同样推送 这时候 电充满了 告诉用户可以走了
            sendChargingFullMessage2User(cost.getUserId(),"你已经充满了,请拔枪走人");
            //sendChargingEnd websocket推送出去
            //修改一下订单状态 为已完成.
            //update charging_bill_success set bill_status=2 where bill_id=
            mqttContext.getBillSuccessMapper().updateBillStatus(cpmd.getBillId(),2);
        }
    }

    private void sendChargingFullMessage2User(Integer userId, String message) {
        mqttContext.getWebSocketServer().pushMessage(userId,message);
    }

    private void sendChargingProcess2User(ChargingProcessVO cost) {
        //消费者使用的MqttContext的桥梁 将容器中bean对象 嫁接过来的
        log.info("即将把充电数据同步到用户手机端:{}",cost);
        //消息转化成json
        String processJson = JSON.toJSONString(cost);
        WebSocketServer server = mqttContext.getWebSocketServer();
        server.pushMessage(cost.getUserId(),processJson);
    }

    private void saveChargingProcessEntity(ChargingProcessMessageDto cpmd, ChargingProcessVO cost) {
        log.info("开始保存充电数据到es,消息入参:{},计价入参",cpmd,cost);
        ChargingProcessEntity entity=new ChargingProcessEntity();
        //提供一些属性封装 大部分属性来自于cpmd
        BeanUtils.copyProperties(cpmd,entity);
        //缺少id 时序id
        entity.setId(System.currentTimeMillis());
        //计算总金额
        entity.setTotalCost(cost.getTotalCost());
        mqttContext.getChargingProcessEntityRepository().save(entity);//存储到es
        log.info("进度记录数据保存成功:{}",entity);
    }
    private ChargingProcessVO billPriceCalc(ChargingProcessMessageDto cpmd) {
        //通过远程调用 给我反馈一个计价的结果 将costClient放到MmqttContext
        CostClient client = mqttContext.getCostClient();
        //准备一个计算价格的入参
        ChargingProcessParam param=new ChargingProcessParam();
        //挨个给准备参数 billId gunId stationId capacity
        BeanUtils.copyProperties(cpmd,param);
        JsonResult<ChargingProcessVO> result = client.costCalc(param);
       /* ChargingProcessVO vo=new ChargingProcessVO();
        vo.setBillId(cpmd.getBillId());
        vo.setUserId(cpmd.getUserId());
        vo.setTotalCost(new BigDecimal(5.0));
        vo.setChargingCapacity(cpmd.getChargingCapacity());
        vo.setGunId(cpmd.getGunId());
        vo.setPowerFee(new BigDecimal(1.5));
        vo.setTotalTime("5小时");*/
        ChargingProcessVO vo = result.getData();
        return vo;
    }




    private ChargingProcessMessageDto parseMessage(MqttMessage message) {
        ChargingProcessMessageDto dto=null;
        if(message!=null){
            //将message的json转化成dto对象
            String jsonMsg = message.toString();
            dto=JSON.parseObject(jsonMsg,ChargingProcessMessageDto.class);
        }
        return dto;
    }

    private void handleChargingResult(MqttMessage message) {
        //1.反序列化 拿到dto消息对象
        ChargingResultMessageDto dto=null;
        //2.message取出字符串数据
        String json = message.toString();
        //3.通过fastjoson工具转化
        dto=JSON.parseObject(json,ChargingResultMessageDto.class);
        //4.简单判断
        if (dto!=null){
            log.info("充电桩结果 消息 :{}",dto);
            if (dto.isSuccess()){
                log.info("充电自检结果成功 billId:{}",dto.getBillId());
                saveSuccessBill(dto);
            }else{
                log.info("充电自检结果失败 billId:{}",dto.getBillId());
                saveFaillBill(dto);
            }
        }
    }

    private void saveFaillBill(ChargingResultMessageDto dto) {
        log.info("设备当前有响应 充电桩充电失败");
        //写入数据库一个BillFailPO
        BillFailPO po=new BillFailPO();
        po.setBillId(dto.getBillId()).setCreateTime(LocalDateTime.now()).setUpdateTime(LocalDateTime.now())
            .setFailDesc("设备自检失败");
        //持久层执行一个写操作 增 删 改 数据库是有反馈的 就是影响的行数(新增一条 影响一行 修改一条影响一行 同时修改全表 影响n行)
        int row = mqttContext.getBillFailMapper().insert(po);
        log.info("保存失败账单记录:{},完成,最终影响行数",po,row);
    }

    private void saveSuccessBill(ChargingResultMessageDto dto) {
        log.info("设备当前有响应 充电桩充电成功");
        //写入一个成功数据
        BillSuccessPO po=new BillSuccessPO();
        //订单状态机 1 充电中 2 充电完成正常结束 3 充电结束 异常结束
        po.setBillId(dto.getBillId()).setCreateTime(LocalDateTime.now()).setUpdateTime(LocalDateTime.now())
            .setBillStatus(OrderStatusConstant.ORDER_STATUS_CHARGING);
        int rows = mqttContext.getBillSuccessMapper().insert(po);
        log.info("保存成功账单记录:{},完成,最终影响行数",po,rows);
    }

    @Override public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("MqttConsumer#deliveryComplete方法被调用了,表示消息分发完成" +
            "token:{}",token);
    }
}
